package com.kygeng.dataupdate.service;

import com.alibaba.fastjson.JSONObject;
import com.kygeng.common.config.ElasticCompanyIndexConfig;

import com.kygeng.common.model.CompanyDetail;
import com.kygeng.common.model.CompanyFullId;
import com.kygeng.common.model.dataconversion.FullDataIdResult;
import com.kygeng.common.model.document.CompanyInfoDoc;
import com.kygeng.common.util.CompanyIndexUtil;
import com.kygeng.dataupdate.mapper.CompanyInfoEsRepository;
import com.kygeng.dataupdate.mapper.CompanyInfoMapper;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * @Title: CompanyDataFullIssuanceService
 * @Author: kygeng
 * @Package: com.kygeng.dataupdate.consumer
 * @Date: 2024/9/8 14:25
 * @Description: company索引全量下发
 */
@Slf4j
@Service
public class CompanyDataFullIssuanceService {

    @Autowired
    private CompanyInfoMapper companyInfoMapper;

    @Autowired
    private CompanyInfoEsRepository companyInfoEsRepository;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    private int batchSize = 8000;//每批次最多获取的数据量
    private int bulkSize = 500;//每次批量插入的数量
    private int threadCount = 5;

    /**
     * 全量下发 company索引相关数据
     */
    public void companyIndexFullIssuance() {

        //1、TODO：暂停增量

        //2、获取company数量
        long startTime = System.currentTimeMillis();
        int totalCount = 0;

        Semaphore semaphore = new Semaphore(threadCount);

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(threadCount, threadCount * 2, 60, TimeUnit.SECONDS, new SynchronousQueue<>());

        int startSeqId = 0;

        while (true) {

            //分批次获取数据
            //全量ID下发+消费
            FullDataIdResult fullDataIdResult = getCompanyInfo(startSeqId, batchSize);
            List<CompanyInfoDoc> companyDataList = (List<CompanyInfoDoc>) fullDataIdResult.getDataList();
            Integer lastId = fullDataIdResult.getLastId();

            if (CollectionUtils.isEmpty(companyDataList)) {
                break;
            } else {
                startSeqId = lastId;
            }

            //拆分为多个 bulk
            int companyListTotal = companyDataList.size();
            int bulkTimes = (int) Math.ceil(companyListTotal / bulkSize);
            totalCount = totalCount + companyListTotal;

            for (int j = 0; j < bulkTimes; j++) {

                List<CompanyInfoDoc> companyData = companyDataList.subList(j * bulkSize, Math.min((j + 1) * bulkSize, companyListTotal));

                semaphore.acquireUninterruptibly();
                threadPoolExecutor.submit(() -> {
                    try {

                        /*BulkRequest bulkRequest = new BulkRequest();
                        for (CompanyInfoDoc companyInfoDoc : companyData) {
                            // 将 CompanyInfoDoc 对象转换为 JSON 字符串
                            String jsonString = JSONObject.toJSONString(companyInfoDoc);

                            // 构建 IndexRequest
                            IndexRequest indexRequest = new IndexRequest(ElasticCompanyIndexConfig.COMPANY_INDEX_NAME)
                                    .id(String.valueOf(companyInfoDoc.getId())) // 设置文档 ID
                                    .source(jsonString, XContentType.JSON);

                            // 将请求添加到 BulkRequest 中
                            bulkRequest.add(indexRequest);
                        }

                        // 执行批量操作
                        BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                        log.info("bulkResponse:", bulkResponse.toString());*/

                        companyInfoEsRepository.saveAll(companyData);

                        log.info("线程[{}]插入[{}]条商品数据", Thread.currentThread().getName(), companyData.size());

                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally {
                        semaphore.release();
                    }
                });
            }
        }

        long endTime = System.currentTimeMillis();

        threadPoolExecutor.shutdown();

        long elapsedSeconds = (endTime - startTime) / 1000;
        long perSecond = totalCount / elapsedSeconds;
        log.info("此次共导入[{}]条商品数据，耗时[{}]秒，平均每秒导入[{}]条数据", totalCount, elapsedSeconds, perSecond);

        //TODO：重新开启增量
    }


    /**
     * 获取信息
     *
     * @param start
     * @param limit
     * @return
     */
    public FullDataIdResult getCompanyInfo(Integer start, Integer limit) {

        //根据id查找companyCode
        //相当于：全量ID下发报表
        List<CompanyFullId> companyFullIdList = companyInfoMapper.selectCompanyCodePage(start, limit);
        if (CollectionUtils.isEmpty(companyFullIdList)) {
            return new FullDataIdResult();
        }

        List<String> companyCodeList = companyFullIdList.stream().map(CompanyFullId::getCompanyCode).collect(Collectors.toList());

        //根据companyCode获取详情
        //相当于：消费报表
        List<CompanyDetail> companyDetails = companyInfoMapper.selectCompanyDetail(companyCodeList);

        //CompanyDetail 转 CompanyInfoDoc
        List<CompanyInfoDoc> companyInfoDocList = CompanyIndexUtil.companyDetailToDoc(companyDetails);

        //定位到本次查询的最后一条数据
        CompanyFullId companyInfoDoc = companyFullIdList.get(companyFullIdList.size() - 1);
        Integer lastSeqId = companyInfoDoc.getId();

        FullDataIdResult result = new FullDataIdResult();
        result.setDataList(companyInfoDocList);
        result.setLastId(lastSeqId);
        return result;
    }
}
